package com.huanke.iot.gateway.mqttlistener;

import com.alibaba.fastjson.JSON;
import com.huanke.iot.base.dao.device.DeviceMapper;
import com.huanke.iot.base.po.device.DevicePo;
import com.huanke.iot.base.po.device.data.DeviceSensorPo;
import com.huanke.iot.gateway.io.AbstractHandler;
import com.huanke.iot.gateway.io.impl.SensorHandler;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Repository;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Repository
@Slf4j
public class MqttService {
    @Autowired
    private DeviceMapper deviceMapper;
    private static String[] huankeDevice = new String[]{"/up2/sensor/2","/up2/sensor/9","/up2/sensor/12","/up2/sensor/13","/up2/sensor/66","/up2/sensor/67","/up2/sensor/68","/up2/sensor/69","/up2/sensor/70","/up2/sensor/71","/up2/sensor/72","/up2/sensor/73","/up2/sensor/74","/up2/sensor/75","/up2/sensor/76","/up2/sensor/77","/up2/sensor/78","/up2/sensor/79","/up2/sensor/80","/up2/sensor/81","/up2/sensor/82","/up2/sensor/83","/up2/sensor/84","/up2/sensor/85","/up2/sensor/86","/up2/sensor/87","/up2/sensor/88","/up2/sensor/89","/up2/sensor/90","/up2/sensor/91","/up2/sensor/92","/up2/sensor/93","/up2/sensor/94","/up2/sensor/95","/up2/sensor/96","/up2/sensor/97","/up2/sensor/99","/up2/sensor/100","/up2/sensor/101","/up2/sensor/102","/up2/sensor/103","/up2/sensor/104","/up2/sensor/105","/up2/sensor/106","/up2/sensor/107","/up2/sensor/108","/up2/sensor/109","/up2/sensor/110","/up2/sensor/111","/up2/sensor/112","/up2/sensor/113","/up2/sensor/114","/up2/sensor/115","/up2/sensor/116","/up2/sensor/117","/up2/sensor/118","/up2/sensor/119","/up2/sensor/120","/up2/sensor/121","/up2/sensor/122","/up2/sensor/123","/up2/sensor/124","/up2/sensor/125","/up2/sensor/126","/up2/sensor/127","/up2/sensor/128","/up2/sensor/129","/up2/sensor/130","/up2/sensor/131","/up2/sensor/132","/up2/sensor/133","/up2/sensor/134","/up2/sensor/166","/up2/sensor/167","/up2/sensor/207","/up2/sensor/209","/up2/sensor/210","/up2/sensor/211","/up2/sensor/212","/up2/sensor/213","/up2/sensor/217","/up2/sensor/218","/up2/sensor/219","/up2/sensor/220","/up2/sensor/221","/up2/sensor/222","/up2/sensor/223","/up2/sensor/224","/up2/sensor/225","/up2/sensor/226","/up2/sensor/227","/up2/sensor/228","/up2/sensor/229","/up2/sensor/230","/up2/sensor/231","/up2/sensor/232","/up2/sensor/233","/up2/sensor/234","/up2/sensor/235","/up2/sensor/236","/up2/sensor/237","/up2/sensor/238","/up2/sensor/239","/up2/sensor/240","/up2/sensor/242","/up2/sensor/243","/up2/sensor/244","/up2/sensor/245","/up2/sensor/246","/up2/sensor/281","/up2/sensor/299","/up2/sensor/300","/up2/sensor/301","/up2/sensor/302","/up2/sensor/303","/up2/sensor/304","/up2/sensor/305","/up2/sensor/306","/up2/sensor/310","/up2/sensor/311","/up2/sensor/312","/up2/sensor/313","/up2/sensor/314","/up2/sensor/315","/up2/sensor/343","/up2/sensor/344","/up2/sensor/345","/up2/sensor/346","/up2/sensor/347","/up2/sensor/348","/up2/sensor/349","/up2/sensor/350","/up2/sensor/357","/up2/sensor/360","/up2/sensor/361","/up2/sensor/362","/up2/sensor/363","/up2/sensor/480","/up2/sensor/481","/up2/sensor/482","/up2/sensor/483","/up2/sensor/484","/up2/sensor/485","/up2/sensor/487","/up2/sensor/488","/up2/sensor/489","/up2/sensor/490","/up2/sensor/491","/up2/sensor/492","/up2/sensor/493","/up2/sensor/494","/up2/sensor/495","/up2/sensor/496","/up2/sensor/497","/up2/sensor/509","/up2/sensor/510","/up2/sensor/511","/up2/sensor/512","/up2/sensor/513","/up2/sensor/514","/up2/sensor/515","/up2/sensor/516","/up2/sensor/517","/up2/sensor/518","/up2/sensor/519","/up2/sensor/520","/up2/sensor/521","/up2/sensor/522","/up2/sensor/523","/up2/sensor/524","/up2/sensor/525","/up2/sensor/526","/up2/sensor/527","/up2/sensor/528","/up2/sensor/529","/up2/sensor/530","/up2/sensor/531","/up2/sensor/533","/up2/sensor/534","/up2/sensor/535","/up2/sensor/536","/up2/sensor/537","/up2/sensor/538","/up2/sensor/539","/up2/sensor/540","/up2/sensor/541","/up2/sensor/542","/up2/sensor/543","/up2/sensor/544","/up2/sensor/545","/up2/sensor/546","/up2/sensor/547","/up2/sensor/548","/up2/sensor/549","/up2/sensor/550","/up2/sensor/551","/up2/sensor/552","/up2/sensor/553","/up2/sensor/554","/up2/sensor/555","/up2/sensor/556","/up2/sensor/557","/up2/sensor/558","/up2/sensor/559","/up2/sensor/560","/up2/sensor/561","/up2/sensor/562","/up2/sensor/563","/up2/sensor/564","/up2/sensor/565","/up2/sensor/566","/up2/sensor/567","/up2/sensor/568","/up2/sensor/569","/up2/sensor/570","/up2/sensor/571","/up2/sensor/572","/up2/sensor/573","/up2/sensor/574","/up2/sensor/575","/up2/sensor/576","/up2/sensor/577","/up2/sensor/578","/up2/sensor/579","/up2/sensor/580","/up2/sensor/581","/up2/sensor/582","/up2/sensor/583","/up2/sensor/584","/up2/sensor/585","/up2/sensor/586","/up2/sensor/587","/up2/sensor/588","/up2/sensor/589","/up2/sensor/590","/up2/sensor/591","/up2/sensor/592","/up2/sensor/593","/up2/sensor/594","/up2/sensor/595","/up2/sensor/596","/up2/sensor/597","/up2/sensor/598","/up2/sensor/599","/up2/sensor/600","/up2/sensor/601","/up2/sensor/602","/up2/sensor/611","/up2/sensor/613","/up2/sensor/614","/up2/sensor/617","/up2/sensor/621","/up2/sensor/622","/up2/sensor/623","/up2/sensor/625","/up2/sensor/2007","/up2/sensor/2008","/up2/sensor/2010","/up2/sensor/2012","/up2/sensor/2013","/up2/sensor/2015","/up2/sensor/2016","/up2/sensor/2017","/up2/sensor/2018","/up2/sensor/2019","/up2/sensor/2020","/up2/sensor/2021","/up2/sensor/2022","/up2/sensor/2023","/up2/sensor/2025","/up2/sensor/2026","/up2/sensor/2027","/up2/sensor/2028","/up2/sensor/2029","/up2/sensor/2031","/up2/sensor/2032","/up2/sensor/2033","/up2/sensor/2034","/up2/sensor/2035","/up2/sensor/2036","/up2/sensor/2037","/up2/sensor/2038","/up2/sensor/2040","/up2/sensor/2041","/up2/sensor/2043","/up2/sensor/2044","/up2/sensor/2046","/up2/sensor/2047","/up2/sensor/2048","/up2/sensor/2049","/up2/sensor/2050","/up2/sensor/2051","/up2/sensor/2053","/up2/sensor/2054","/up2/sensor/2055","/up2/sensor/2057","/up2/sensor/2058","/up2/sensor/2061","/up2/sensor/2062","/up2/sensor/2063","/up2/sensor/2064","/up2/sensor/2065","/up2/sensor/2066","/up2/sensor/2067","/up2/sensor/2068","/up2/sensor/2069","/up2/sensor/2070","/up2/sensor/2072","/up2/sensor/2073","/up2/sensor/2074","/up2/sensor/2076","/up2/sensor/2077","/up2/sensor/2078","/up2/sensor/2079","/up2/sensor/2080","/up2/sensor/2081","/up2/sensor/2082","/up2/sensor/2083","/up2/sensor/2084","/up2/sensor/2085","/up2/sensor/2086","/up2/sensor/2087","/up2/sensor/2088","/up2/sensor/2089","/up2/sensor/2090","/up2/sensor/2091","/up2/sensor/2093","/up2/sensor/2094","/up2/sensor/2095","/up2/sensor/2096","/up2/sensor/2097","/up2/sensor/2098","/up2/sensor/2099","/up2/sensor/2100","/up2/sensor/2101","/up2/sensor/2102","/up2/sensor/2103","/up2/sensor/2104","/up2/sensor/2105","/up2/sensor/2108","/up2/sensor/2109","/up2/sensor/2110","/up2/sensor/2111","/up2/sensor/2113","/up2/sensor/2114","/up2/sensor/2115","/up2/sensor/2116","/up2/sensor/2117","/up2/sensor/2118","/up2/sensor/2119","/up2/sensor/2120","/up2/sensor/2121","/up2/sensor/2122","/up2/sensor/2123","/up2/sensor/2124","/up2/sensor/2126","/up2/sensor/2128","/up2/sensor/2129","/up2/sensor/2130","/up2/sensor/2131","/up2/sensor/2132","/up2/sensor/2133","/up2/sensor/2134","/up2/sensor/2135","/up2/sensor/2136","/up2/sensor/2137","/up2/sensor/2138","/up2/sensor/2139","/up2/sensor/2140","/up2/sensor/2141","/up2/sensor/2142","/up2/sensor/2143","/up2/sensor/2144","/up2/sensor/2145","/up2/sensor/2146","/up2/sensor/2147","/up2/sensor/2148","/up2/sensor/2149","/up2/sensor/2150","/up2/sensor/2151","/up2/sensor/2152","/up2/sensor/2153","/up2/sensor/2154","/up2/sensor/2155","/up2/sensor/2156","/up2/sensor/2157","/up2/sensor/2158","/up2/sensor/2159","/up2/sensor/2160","/up2/sensor/2161","/up2/sensor/2163","/up2/sensor/2164","/up2/sensor/2165","/up2/sensor/2166","/up2/sensor/2167","/up2/sensor/2168","/up2/sensor/2169","/up2/sensor/2170","/up2/sensor/2171","/up2/sensor/2172","/up2/sensor/2173","/up2/sensor/2174","/up2/sensor/2176","/up2/sensor/2178","/up2/sensor/2179","/up2/sensor/2181","/up2/sensor/2182","/up2/sensor/2183","/up2/sensor/2184","/up2/sensor/2185","/up2/sensor/2186","/up2/sensor/2187","/up2/sensor/2188","/up2/sensor/2189","/up2/sensor/2190","/up2/sensor/2191","/up2/sensor/2192","/up2/sensor/2193","/up2/sensor/2194","/up2/sensor/2195","/up2/sensor/2196","/up2/sensor/2198","/up2/sensor/2199","/up2/sensor/2200","/up2/sensor/2201","/up2/sensor/2202","/up2/sensor/2203","/up2/sensor/2204","/up2/sensor/2205","/up2/sensor/2206","/up2/sensor/2207","/up2/sensor/2208","/up2/sensor/2209","/up2/sensor/2211","/up2/sensor/2212","/up2/sensor/2215","/up2/sensor/2216","/up2/sensor/2217","/up2/sensor/2218","/up2/sensor/2219","/up2/sensor/2220","/up2/sensor/2221","/up2/sensor/2222","/up2/sensor/2223","/up2/sensor/2224","/up2/sensor/2225","/up2/sensor/2226","/up2/sensor/2227","/up2/sensor/2228","/up2/sensor/2229","/up2/sensor/2230","/up2/sensor/2231","/up2/sensor/2232","/up2/sensor/2233","/up2/sensor/2234","/up2/sensor/2236","/up2/sensor/2237","/up2/sensor/2238","/up2/sensor/2239","/up2/sensor/2240","/up2/sensor/2241","/up2/sensor/2242","/up2/sensor/2243","/up2/sensor/2244","/up2/sensor/2245","/up2/sensor/2246","/up2/sensor/2247","/up2/sensor/2248","/up2/sensor/2249","/up2/sensor/2250","/up2/sensor/2251","/up2/sensor/2252","/up2/sensor/2253","/up2/sensor/2254","/up2/sensor/2256","/up2/sensor/2257","/up2/sensor/2258","/up2/sensor/2259","/up2/sensor/2260","/up2/sensor/2261","/up2/sensor/2262","/up2/sensor/2264","/up2/sensor/2265","/up2/sensor/2266","/up2/sensor/2268","/up2/sensor/2269","/up2/sensor/2270","/up2/sensor/2271","/up2/sensor/2273","/up2/sensor/2274","/up2/sensor/2275","/up2/sensor/2276","/up2/sensor/2277","/up2/sensor/2278","/up2/sensor/2279","/up2/sensor/2280","/up2/sensor/2281","/up2/sensor/2282","/up2/sensor/2283","/up2/sensor/2284","/up2/sensor/2285","/up2/sensor/2286","/up2/sensor/2287","/up2/sensor/2288","/up2/sensor/2289","/up2/sensor/2290","/up2/sensor/2291","/up2/sensor/2292","/up2/sensor/2294","/up2/sensor/2295","/up2/sensor/2296","/up2/sensor/2297","/up2/sensor/2298","/up2/sensor/2299","/up2/sensor/2300","/up2/sensor/2301","/up2/sensor/2302","/up2/sensor/2303","/up2/sensor/2305","/up2/sensor/2306","/up2/sensor/2308","/up2/sensor/2309","/up2/sensor/2312","/up2/sensor/2313","/up2/sensor/2314","/up2/sensor/2315","/up2/sensor/2316","/up2/sensor/2317","/up2/sensor/2318","/up2/sensor/2319","/up2/sensor/2320","/up2/sensor/2321","/up2/sensor/2322","/up2/sensor/2323","/up2/sensor/2324","/up2/sensor/2325","/up2/sensor/2326","/up2/sensor/2327","/up2/sensor/2328","/up2/sensor/2329","/up2/sensor/2330","/up2/sensor/2331","/up2/sensor/2332","/up2/sensor/2333","/up2/sensor/2334","/up2/sensor/2335","/up2/sensor/2336","/up2/sensor/2337","/up2/sensor/2338","/up2/sensor/2339","/up2/sensor/2341","/up2/sensor/2342","/up2/sensor/2343","/up2/sensor/2344","/up2/sensor/2345","/up2/sensor/2346","/up2/sensor/2347","/up2/sensor/2348","/up2/sensor/2349","/up2/sensor/2350","/up2/sensor/2351","/up2/sensor/2352","/up2/sensor/2353","/up2/sensor/2354","/up2/sensor/2355","/up2/sensor/2356","/up2/sensor/2357","/up2/sensor/2358","/up2/sensor/2359","/up2/sensor/2360","/up2/sensor/2361","/up2/sensor/2362","/up2/sensor/2363","/up2/sensor/2364","/up2/sensor/2365","/up2/sensor/2366","/up2/sensor/2367","/up2/sensor/2368","/up2/sensor/2369","/up2/sensor/2370","/up2/sensor/2371","/up2/sensor/2375","/up2/sensor/2376","/up2/sensor/2377","/up2/sensor/2378","/up2/sensor/2379","/up2/sensor/2380","/up2/sensor/2381","/up2/sensor/2382","/up2/sensor/2383","/up2/sensor/2384","/up2/sensor/2385","/up2/sensor/2386","/up2/sensor/2387","/up2/sensor/2388","/up2/sensor/2389","/up2/sensor/2391","/up2/sensor/2392","/up2/sensor/2393","/up2/sensor/2394","/up2/sensor/2395","/up2/sensor/2396","/up2/sensor/2397","/up2/sensor/2398","/up2/sensor/2399","/up2/sensor/2400","/up2/sensor/2401","/up2/sensor/2402","/up2/sensor/2403","/up2/sensor/2405","/up2/sensor/2406","/up2/sensor/2407","/up2/sensor/2408","/up2/sensor/2409","/up2/sensor/2410","/up2/sensor/2411","/up2/sensor/2412","/up2/sensor/2413","/up2/sensor/2414","/up2/sensor/2415","/up2/sensor/2416","/up2/sensor/2417","/up2/sensor/2418","/up2/sensor/2419","/up2/sensor/2420","/up2/sensor/2421","/up2/sensor/2422","/up2/sensor/2423","/up2/sensor/2425","/up2/sensor/2426","/up2/sensor/2427","/up2/sensor/2429","/up2/sensor/2430","/up2/sensor/2431","/up2/sensor/2432","/up2/sensor/2433","/up2/sensor/2435","/up2/sensor/2436","/up2/sensor/2437","/up2/sensor/2439","/up2/sensor/2440","/up2/sensor/2441","/up2/sensor/2443","/up2/sensor/2444","/up2/sensor/2445","/up2/sensor/2446","/up2/sensor/2447","/up2/sensor/2448","/up2/sensor/2449","/up2/sensor/2450","/up2/sensor/2451","/up2/sensor/2453","/up2/sensor/2454","/up2/sensor/2455","/up2/sensor/2469","/up2/sensor/2474","/up2/sensor/2475","/up2/sensor/2476","/up2/sensor/2480","/up2/sensor/100002","/up2/sensor/100004","/up2/sensor/100218"};
//    private static String[] huankeDevice = new String[]{"/up2/sensor/2110"};

    private static Map<String, Long> noDealMap = new HashMap<>();


    @Value("${mqttOldServerUrl}")
    private String mqttOldServerUrl;

    @Value("${mqttNewServerUrl}")
    private String mqttNewServerUrl;

    MqttClient oldServerClient;

    MqttClient newServerClient;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @PostConstruct
    public void init(){
        if(oldServerClient == null){
            try {
                oldServerClient = new MqttClient(mqttOldServerUrl, "ServerClientNew1");
            }catch (Exception e){
                log.error("",e);
            }
        }

        if(newServerClient == null){
            try {
                newServerClient = new MqttClient(mqttNewServerUrl, "ServerClientNew2");
            }catch (Exception e){
                log.error("",e);
            }
        }
    }
    @EventListener
    public void start(ApplicationReadyEvent  event){
        if(oldServerClient != null){
            try {
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true);
                connOpts.setAutomaticReconnect(true);
                oldServerClient.connect(connOpts);
                MqttMessageListener listener = new MqttMessageListener();
                oldServerClient.subscribe("/up2/#",listener);
                //监听遗言
                oldServerClient.subscribe("/lastwill/#",listener);
//                oldServerClient.subscribe(huankeDevice, new int[]{1}, new IMqttMessageListener[]{listener});
            }catch (Exception e){
                log.error("",e);
            }
        }

        if(newServerClient != null){
            try {
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true);
                connOpts.setAutomaticReconnect(true);
                newServerClient.connect(connOpts);
                MqttMessageListener listener = new MqttMessageListener();
                newServerClient.subscribe("/up2/#",listener);
                //监听遗言
                newServerClient.subscribe("/lastwill/#",listener);
            }catch (Exception e){
                log.error("",e);
            }
        }
    }

    public class MqttMessageListener implements IMqttMessageListener{

        DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup(240,new DefaultThreadFactory("MqttMessageThread"));
        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            String deviceId = topic.substring(topic.lastIndexOf("/")+1);
            if(deviceId.equals("2110")){
                log.info("topic -----> {},content->{}",topic, new String(message.getPayload()));
            }

            if(topic.startsWith("/up2/sensor")){
                if(noDealMap.get(deviceId) != null && System.currentTimeMillis() - noDealMap.get(deviceId) < 6000){
                    return;
                }
                //标记设备不读取缓存
                noDealMap.put(deviceId, System.currentTimeMillis());
                SensorHandler.SensorListMessage sensorListMessage = JSON.parseObject(new String(message.getPayload()),SensorHandler.SensorListMessage.class);

                List<SensorHandler.SensorMessage> datasPower = new ArrayList<>();
                sensorListMessage.getDatas().forEach(sensorMessage -> {
                    int deviceId1 = Integer.valueOf(topic.substring(topic.lastIndexOf("/") + 1));
                    if (sensorMessage.getChildid() != null && !sensorMessage.getChildid().equals("0")) {
                        DevicePo childDevice = deviceMapper.getChildDevice(deviceId1, sensorMessage.getChildid());
                        if (childDevice != null) {
                            deviceId1 = childDevice.getId();
                        } else {
                            return;
                        } }DeviceSensorPo deviceSensorPo = new DeviceSensorPo();
                    String type = sensorMessage.getType();
                    if (StringUtils.isBlank(type)) {
                        type = sensorMessage.getT();
                    }
                    deviceSensorPo.setSensorType(type);
                    String value = sensorMessage.getValue();
                    if (value == null) {
                        value = sensorMessage.getV();
                    }
                    deviceSensorPo.setSensorValue(value);
                    deviceSensorPo.setCreateTime(System.currentTimeMillis());
                    stringRedisTemplate.opsForHash().put("sensor2." + deviceId1, type, String.valueOf(value));
                    //stringRedisTemplate.expire("sensor2." + deviceId1,30, TimeUnit.MINUTES);
                });
            }



            defaultEventLoopGroup.submit(new Runnable() {
                @Override
                public void run() {
                    AbstractHandler.getHandler(topic).handler(topic,message.getPayload());
                }
            });
        }
    }

    public void sendMessage(String topic,String message,boolean isOld){
        MqttClient mqttClient = getClient(isOld);
        if(mqttClient != null){
            try {
                mqttClient.publish(topic, new MqttMessage(message.getBytes()));
            }catch (Exception e){
                log.error("",e);
            }
        }
    }

    private MqttClient getClient(boolean isOld){
        MqttClient mqttClient = oldServerClient;
        if(!isOld){
            mqttClient = newServerClient;
        }
        return mqttClient;
    }
}
